package com.shujia.deploy

import java.sql.{Connection, PreparedStatement, ResultSet}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.shujia.card.RealTimeCardWindowFlow.env
import com.shujia.flink.FlinkTool
import com.shujia.util.{CarUtil, JdbcUtil}
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.DataStream

import scala.collection.mutable.ListBuffer

/**
  * 事实稽查布控
  */
object HaveBeenSupervisedCheck extends FlinkTool {
  override def run(args: Array[String]): Unit = {
    //读取数据
    val carDS: DataStream[String] = CarUtil.loadKafkaData(env)


    /**
      * 处理过车数据，实时读取布控列表，和过车数据进行对比
      *
      */


    carDS.addSink(new RichSinkFunction[String] {


      override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {
        //读取布控列表
        val stat: PreparedStatement = con.prepareStatement("select car from deploy_list")
        val resultSet: ResultSet = stat.executeQuery()
        //布控列表
        val cars = new ListBuffer[String]
        while (resultSet.next()) {
          val car: String = resultSet.getString("car")
          cars += car
        }

        //获取数据中的车牌号
        val jsonObj: JSONObject = JSON.parseObject(value)
        val car: String = jsonObj.getString("car")

        //如果布控列表中的车辆出现了将车辆的数据保存到mysql
        if (cars.contains(car)) {

          //{"car":"皖ANQEQ9",
          // "city_code":"340500",
          // "county_code":"340522",
          // "card":118702031654010,
          // "camera_id":"00234",
          // "orientation":"东",
          // "road_id":34051981,
          // "time":1614726264,"speed":89.89}
          val car: String = jsonObj.getString("car")
          val city_code: String = jsonObj.getString("city_code")
          val county_code: String = jsonObj.getString("county_code")
          val card: Long = jsonObj.getLong("card")
          val camera_id: String = jsonObj.getString("camera_id")
          val orientation: String = jsonObj.getString("orientation")
          val road_id: Long = jsonObj.getLong("road_id")
          val time: Long = jsonObj.getLong("time")
          val speed: Double = jsonObj.getDouble("speed")


          val stat: PreparedStatement = con.prepareStatement(
            """
              |replace into deploy_data(
              |car,
              |city_code,
              |county_code,
              |card,
              |camera_id,
              |orientation,
              |road_id,
              |time,
              |speed
              |)
              |values(?,?,?,?,?,?,?,?,?)
              |
            """.stripMargin)

          stat.setString(1, car)
          stat.setString(2, city_code)
          stat.setString(3, county_code)
          stat.setLong(4, card)
          stat.setString(5, camera_id)
          stat.setString(6, orientation)
          stat.setLong(7, road_id)
          stat.setLong(8, time)
          stat.setDouble(9, speed)

          stat.execute()

        }
      }

      var con: Connection = _

      /**
        * 在filter 之前执行。。每一个task中只执行一次
        *
        */
      override def open(parameters: Configuration): Unit = {
        con = JdbcUtil.getConnection
      }

      override def close(): Unit = {
        con.close()
      }

    })

    env.execute("HaveBeenSupervisedCheck")


  }
}
